package avicit.bdp.dds.server.worker.task.algorithm;

import avicit.bdp.common.datasource.BaseDataSource;
import avicit.bdp.common.datasource.BaseKerberosDataSource;
import avicit.bdp.common.datasource.MySQLDataSource;
import avicit.bdp.common.utils.DESUtils;
import avicit.bdp.dds.dispatch.process.Property;
import avicit.bdp.dds.dispatch.task.algorithm.AlgorithmParameters;
import avicit.bdp.core.util.json.JSONUtils;
import avicit.bdp.dds.common.utils.FileUtils;
import avicit.bdp.dds.server.entity.TaskExecutionContext;
import avicit.bdp.dds.service.process.ProcessService;
import com.alibaba.fastjson2.JSON;
import com.alibaba.fastjson2.JSONObject;
import java.io.File;
import java.io.IOException;
import java.nio.charset.StandardCharsets;
import java.nio.file.Files;
import java.nio.file.Path;
import java.util.HashMap;
import java.util.Map;
import java.util.stream.Collectors;
import org.apache.commons.collections4.MapUtils;
import org.apache.commons.lang3.StringUtils;
import org.slf4j.Logger;

/**
 *
 *
 * @author xugb
 * @date 2021/1/18 16:34
 */
public class InputJsonUtils {
    /**
     * 生成inputJson文件
     *
     * @param taskExecutionContext
     * @param logger
     * @param algorithmParameters
     * @param processService
     * @return java.lang.String
     * @author xugb
     * @date 2021/1/18 16:43
     */
    public static String buildAlgoJsonFile(TaskExecutionContext taskExecutionContext, Logger logger,
                                           AlgorithmParameters algorithmParameters,
                                           ProcessService processService) throws IOException {
        // generate json
        String fileName = String.format("%s/%s_job.json",
                taskExecutionContext.getExecutePath(), taskExecutionContext.getTaskAppId());
        Path path = new File(fileName).toPath();
        if (Files.exists(path)) {
            return fileName;
        }

        JSONObject root = new JSONObject();
        root.put("name", algorithmParameters.getNodeId());
        root.put("parameter", buildParameterJson(taskExecutionContext, algorithmParameters, processService));
        String json = root.toString();
        logger.info("algo job json : {}", json);

        // create  json file
        FileUtils.writeStringToFile(new File(fileName), json, StandardCharsets.UTF_8);

        return fileName;
    }

    /**
     * 生成json参数
     *
     * @return json参数
     */
    private static JSONObject buildParameterJson(TaskExecutionContext taskExecutionContext, AlgorithmParameters algorithmParameters, ProcessService processService) {
        JSONObject param = new JSONObject();

        param.put("type", algorithmParameters.getType());
        param.put("input", handleIOParams(algorithmParameters.getInputParams(), processService));
        param.put("output", handleIOParams(algorithmParameters.getOutputParams(), processService));
        param.put("settingParams", handleSettingParams(algorithmParameters.getSettingParams()));


        // 全局参数和自定义参数
        Map<String, String> userDefParamsMaps = new HashMap<>(16);
        userDefParamsMaps.putAll(taskExecutionContext.getDefinedParams());
        if (algorithmParameters.getLocalParametersMap() != null) {
            userDefParamsMaps.putAll(algorithmParameters.getLocalParametersMap().values()
                    .stream()
                    .collect(Collectors.toMap(Property::getProp, Property::getValue)));
        }
        param.put("userDefParams", JSON.parseObject(JSON.toJSONString(userDefParamsMaps)));

        return param;
    }

    private static JSONObject handleSettingParams(String settingParams) {
        JSONObject params = new JSONObject();

        if (StringUtils.isNotBlank(settingParams)) {
            params = JSON.parseObject(settingParams);
        }

        return params;
    }

    private static JSONObject handleIOParams(String ioParams, ProcessService processService) {
        JSONObject params = new JSONObject();

        Map<String, Object> input = JSONUtils.toObjectMap(ioParams);
        //输入类型：0-数据源、1-文件、2-数据模型、3-目录、-1-什么都不选
        Integer inputType = MapUtils.getInteger(input, "inputType", -1);
        if (inputType == null) {
            inputType = -1;
        }
        params.put("inputType", inputType);
        params.put("saveMode", MapUtils.getString(input, "saveMode", "overwrite"));//"append/overwrite",--存储模式：添加、覆盖
        if (2 == inputType) {
            String tableId = MapUtils.getString(input, "dbId", "");
            if (StringUtils.isNotBlank(tableId)) {
                String inputDatasourceId = processService.getDataSourceIdByTableId(tableId);
                BaseDataSource dataSourceCfg = processService.getDataSource(inputDatasourceId);
                if (dataSourceCfg == null) {
                    dataSourceCfg = new MySQLDataSource();
                }
                handleConnectionInfo(params, dataSourceCfg);
                String table = processService.getTableNameByTableId(tableId);
                params.put("table", table);
            }
        } else if (1 == inputType) {
            //存放文件  TODO  支持其他文件存储方式，文件连接路径
            params.put("dbType", "hdfs");
            params.put("resourceId", MapUtils.getString(input, "resourceId"));
            params.put("fileFormat", MapUtils.getString(input, "fileFormat"));
            params.put("filePath", MapUtils.getString(input, "filePath"));
            params.put("fileHeader", MapUtils.getBoolean(input, "fileHeader", true));
            params.put("fileDelimiter", MapUtils.getString(input, "fileDelimiter", ","));
        } else if (3 == inputType) {
            //目录
            params.put("dbType", "hdfs");
            params.put("filePath", MapUtils.getString(input, "filePath"));
        } else if (0 == inputType) {
            String dbId = MapUtils.getString(input, "dbId", "");
            BaseDataSource dataSourceCfg = processService.getDataSource(dbId);
            if (dataSourceCfg == null) {
                dataSourceCfg = new MySQLDataSource();
            }
            handleConnectionInfo(params, dataSourceCfg);
        } else {

        }
        String fieldParams = MapUtils.getString(input, "fieldParams");
        if (StringUtils.isNotBlank(fieldParams)) {
            params.put("fieldParams", JSONUtils.parseObject(fieldParams));
        } else {
            params.put("fieldParams", "");
        }

        return params;
    }

    /**
     * 处理数据源连接信息
     *
     * @param params
     * @param dataSource
     * @return void
     * @author xugb
     * @date 2021/1/28 15:41
     */
    private static void handleConnectionInfo(JSONObject params, BaseDataSource dataSource) {
        params.put("username", dataSource.getUser());
        params.put("password", DESUtils.decrypt(dataSource.getPassword()));
        params.put("jdbcUrl", dataSource.getJdbcUrl());
        params.put("driver", dataSource.driverClassSelector());
        params.put("dbType", dataSource.type());

        String jdbcJarsPath = FileUtils.getResourceDirectoryFileNames("libs", "jar");
        params.put("jdbcJars", jdbcJarsPath);
        //处理kerberos
        if (dataSource instanceof BaseKerberosDataSource) {
            BaseKerberosDataSource ds = (BaseKerberosDataSource) dataSource;
            params.put("haveKerberos", ds.getKerberos() != null && ds.getKerberos() == 1 ? true : false);
            params.put("krb5Path", ds.getKrb5Path());
            params.put("keytabPath", ds.getKeytabPath());
            params.put("principal", ds.getPrincipal());
        } else {
            params.put("haveKerberos", false);
            params.put("krb5Path", "");
            params.put("keytabPath", "");
            params.put("principal", "");
        }
    }

}
